package rabbitmq

import (
	"fmt"
	"log"

	"github.com/streadway/amqp"
)

// 连接信息 amqp:账户：密码//ip:host/vhost
const MQURL = "amqp://rmqroot:rmqroot@127.0.0.1:5672/go-seckill"
// const MQURL = "amqp://guest:guest@127.0.0.1:5672/"

var (
	FailedConnected   = "failed to connect rabbitmq"
	FailedOpenChannel = "failed to open a channel"
)

type RabbitMQ struct {
	//
	conn *amqp.Connection
	// 通道，生产者/消费者通过通道通信
	channel *amqp.Channel
	// 队列名称   队列收到消息，则将消息发给消费者
	QueueName string
	// 交换机名称  类似路由器，服务器根据路由键将消息从交换器路由到队列上
	Exchange string
	// Key  绑定的key
	Key string
	// 连接信息
	MQURL string
}

// New
func NewRabbitMQ(queueName string, exchange string, key string) *RabbitMQ {
	return &RabbitMQ{
		QueueName: queueName,
		Exchange:  exchange,
		Key:       key,
		MQURL:     MQURL,
	}
}

// 断开连接
func (rmq *RabbitMQ) Destroy() {
	rmq.channel.Close()
	rmq.conn.Close()
}

// 抽象出一个err logger
func (r *RabbitMQ) LogErr(err error, message string) {
	if err != nil {
		errMsg := fmt.Sprintf("%s:%s", message, err)
		log.Println(errMsg)
		panic(errMsg)
	}
}

// RabbitMQ模式
// 工作模式	描述
// simple	最简单的收发模式
// work	资源的竞争
// publish/subscribe	共享资源
// routing	只能匹配上路由key对应的消息队列,对应的消费者才能消费消息
// topic	routing的一种模糊匹配

// 创建简单模式的MQ
func NewRabbitMQSimple(queueName string) *RabbitMQ {
	// 创建一个实例   exchange string, key string  空
	rabbitMQ := NewRabbitMQ(queueName, "", "")
	var err error
	// 获取connection
	rabbitMQ.conn, err = amqp.Dial(rabbitMQ.MQURL)
	rabbitMQ.LogErr(err, FailedConnected)

	// 获取channel
	rabbitMQ.channel, err = rabbitMQ.conn.Channel()
	rabbitMQ.LogErr(err, FailedOpenChannel)

	return rabbitMQ
}

// 简单模式下的队列生产
func (r *RabbitMQ) PublishSimple(msg string) {
	// 生产，将数据写入消息队列
	_, err := r.channel.QueueDeclare(
		r.QueueName, //名称
		false,       // durable 是否持久化
		false,       // autoDelete 自动删除
		false,       //exclusive 是否具有排他性
		false,       //nowWait  是否阻塞处理
		nil,         // args amqb.Table  额外的属性
	)
	if err != nil {
		fmt.Println(err)
	}

	// 写入消息到消息队列
	// Publish(exchange string, key string, mandatory bool, immediate bool, msg amqp.Publishing) error
	r.channel.Publish(
		r.Exchange,
		r.QueueName,
		false, //mandatory 若为true，则根据自身exhcange规则和routeKey规则；无法找到符合条件的队列会把消息返还给发送者
		false, //immediate  如果为true，当exchange发送消息到队列后发现队列上没有消费者，则会把消息返还给发送者
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(msg),
		},
	)

}

// 简单模式下的消费
func (r *RabbitMQ) ConsumeSimple() {
	// 申请队列，若不存在则会创建
	q, err := r.channel.QueueDeclare(
		r.QueueName,
		false, //是否持久化
		false, //是否自动删除
		false, //是否具有排他性
		false, //是否阻塞处理
		nil,   //额外的属性
	)

	if err != nil {
		fmt.Println(err)
	}

	// 接收 消费
	// func (*amqp.Channel).Consume(queue string, consumer string, autoAck bool, exclusive bool, noLocal bool, noWait bool, args amqp.Table) (<-chan amqp.Delivery, error)
	msg, err := r.channel.Consume(
		q.Name, // queue
		"",     //consumer 用来区分多个消费者 此处不区分
		true,   //autoAck 是否自动应答
		false,  //exclusive 是否独有 排他
		false,  //noLocal 设置为true，表示不能将同一个Connection中生产者发送的消息传递给这个Connection中的消费者
		false,  //noWait 是否阻塞处理
		nil,    //args 额外的属性
	)
	if err != nil {
		fmt.Println(err)
	}
	// 处理消息
	// 此处使用forever的意思为因为协程会始终监听消息(除非手动结束)
	// 手动结束才会进行 <-forever 有协程且一直尝试读取数据
	forever := make(chan bool)
	go func() {
		for d := range msg {
			log.Printf("Receive message: %s", d.Body)
		}
	}()
	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	<-forever
}

// NewRabbitMQPubSub 订阅模式创建RabbitMQ实例就要设置路由器了
func NewRabbitMQPubSub(exchangeName string) *RabbitMQ {
	//todo 创建RabbitMQ实例
	rabbitmq := NewRabbitMQ("", exchangeName, "")
	var err error
	//todo 获取connection和获取channel
	rabbitmq.conn, err = amqp.Dial(rabbitmq.MQURL)
	rabbitmq.LogErr(err, "failed to connect rabbitmq!")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.LogErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishPub 订阅模式生产
func (r *RabbitMQ) PublishPub(message string) {
	//todo 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout",
		true,
		false,
		false, //true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		nil,
	)

	r.LogErr(err, "Failed to declare an exchange")

	//todo 发送消息
	err = r.channel.Publish(
		r.Exchange,
		"",
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

// ReceiveSub 订阅模式消费端代码
func (r *RabbitMQ) ReceiveSub() {
	//todo 试探性创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"fanout", //交换机类型
		true,
		false,
		false, //true表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定
		false,
		nil,
	)
	r.LogErr(err, "Failed to declare an exchange")
	//todo 试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", //随机生产队列名称
		false,
		false,
		true,
		false,
		nil,
	)
	r.LogErr(err, "Failed to declare a queue")

	//todo 绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		"", //在pub/sub模式下，这里的key要为空
		r.Exchange,
		false,
		nil)

	//todo 消费消息
	message, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)
	go func() {
		for d := range message {
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	fmt.Println("退出请按 CTRL+C\n")
	<-forever
}

// NewRabbitMQRouting 路由模式
func NewRabbitMQRouting(exchangeName string, routingKey string) *RabbitMQ {
	//todo 创建RabbitMQ实例
	rabbitmq := NewRabbitMQ("", exchangeName, routingKey)
	var err error
	//todo 获取connection 获取channel
	rabbitmq.conn, err = amqp.Dial(rabbitmq.MQURL)
	rabbitmq.LogErr(err, "failed to connect rabbitmq!")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.LogErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishRouting 路由模式发送消息
func (r *RabbitMQ) PublishRouting(message string) {
	//1.尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"direct",
		true,
		false,
		false,
		false,
		nil,
	)

	r.LogErr(err, "Failed to declare an exchange")

	//2.发送消息
	err = r.channel.Publish(
		r.Exchange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

// ReceiveRouting 路由模式接受消息
func (r *RabbitMQ) ReceiveRouting() {
	// todo 试探性创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		//交换机类型
		"direct",
		true,
		false,
		false,
		false,
		nil,
	)
	r.LogErr(err, "Failed to declare an exchange")
	// todo 试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", //随机生产队列名称
		false,
		false,
		true,
		false,
		nil,
	)
	r.LogErr(err, "Failed to declare a queue")

	//绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		//需要绑定key
		r.Key,
		r.Exchange,
		false,
		nil)

	// todo 消费消息
	message, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)
	forever := make(chan bool)
	go func() {
		for d := range message {
			log.Printf("Received a message: %s", d.Body)
		}
	}()
	fmt.Println("退出请按 CTRL+C\n")
	<-forever
}

// NewRabbitMQTopic 话题模式
func NewRabbitMQTopic(exchangeName string, routingKey string) *RabbitMQ {
	// todo 创建RabbitMQ实例
	rabbitmq := NewRabbitMQ("", exchangeName, routingKey)
	var err error
	// todo 获取connection与获取channel
	rabbitmq.conn, err = amqp.Dial(rabbitmq.MQURL)
	rabbitmq.LogErr(err, "failed to connect rabbitmq!")
	rabbitmq.channel, err = rabbitmq.conn.Channel()
	rabbitmq.LogErr(err, "failed to open a channel")
	return rabbitmq
}

// PublishTopic 话题模式发送消息
func (r *RabbitMQ) PublishTopic(message string) {
	// todo 尝试创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		"topic",
		true,
		false,
		false,
		false,
		nil,
	)

	r.LogErr(err, "Failed to declare an exchange")

	// todo 发送消息
	err = r.channel.Publish(
		r.Exchange,
		//要设置
		r.Key,
		false,
		false,
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(message),
		})
}

// ReceiveTopic 话题模式接受消息
//要注意key,规则
//其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）
//匹配 xx.* 表示匹配 xx.hello, 但是 xx.hello.one需要用 xx.#才能匹配到
func (r *RabbitMQ) ReceiveTopic() {
	//1.试探性创建交换机
	err := r.channel.ExchangeDeclare(
		r.Exchange,
		//交换机类型
		"topic",
		true,
		false,
		false,
		false,
		nil,
	)
	r.LogErr(err, "Failed to declare an exch"+
		"ange")
	//2.试探性创建队列，这里注意队列名称不要写
	q, err := r.channel.QueueDeclare(
		"", //随机生产队列名称
		false,
		false,
		true,
		false,
		nil,
	)
	r.LogErr(err, "Failed to declare a queue")

	//绑定队列到 exchange 中
	err = r.channel.QueueBind(
		q.Name,
		//在pub/sub模式下，这里的key要为空
		r.Key,
		r.Exchange,
		false,
		nil)

	//消费消息
	message, err := r.channel.Consume(
		q.Name,
		"",
		true,
		false,
		false,
		false,
		nil,
	)

	forever := make(chan bool)

	go func() {
		for d := range message {
			log.Printf("Received a message: %s", d.Body)
		}
	}()

	fmt.Println("退出请按 CTRL+C\n")
	<-forever
}
